package com.st.jms.kafka;


import com.st.jms.handler.MessageHandler;
import com.st.jms.util.ReadProperties;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerClient extends Thread {

  private String zkConnect;
  private String topic;
  private String groupId;
  private MessageHandler messageHandler;

  private ConsumerConnector consumerConnector;
  private Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap;

  public ConsumerClient(String zkConnect, String topic, String groupId, MessageHandler handler) throws RuntimeException {
    this.zkConnect = zkConnect;
    this.topic = topic;
    this.groupId = groupId;
    this.messageHandler = handler;

    consumerConnector = Consumer.createJavaConsumerConnector(createConsumerConfig());
    consumerConnect();
  }

  private ConsumerConfig createConsumerConfig() {
    Properties props = new Properties();
    props.put("zookeeper.connect", zkConnect);
    props.put("group.id", groupId);
    props.put("zookeeper.connection.timeout.ms", ReadProperties.getPropDefIfBlank("zkConnectionMs", "15000"));
    props.put("zookeeper.session.timeout.ms", ReadProperties.getPropDefIfBlank("zkSessionMs", "400"));
    props.put("zookeeper.sync.time.ms", ReadProperties.getPropDefIfBlank("zkSyncMs", "200"));
    props.put("auto.commit.interval.ms", ReadProperties.getPropDefIfBlank("autoCommitMs", "2000"));
    return new ConsumerConfig(props);
  }

  private void consumerConnect() {
    Map<String, Integer> topicCountMap = new HashMap<>();
    topicCountMap.put(topic, 1);
    consumerMap = consumerConnector.createMessageStreams(topicCountMap);
  }

  @Override
  public void run() {
    try {
      KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
      ConsumerIterator<byte[], byte[]> it = stream.iterator();
      while (it.hasNext() && !Thread.interrupted()) {
        messageHandler.doneMessage(it.next().message());
      }
    } catch (Exception e) {
      e.printStackTrace();
      System.err.println("读kafka队列异常终止！");
      consumerMap.clear();
      consumerConnector.shutdown();
    }
  }
}
